package com.qingcheng.mq;

import com.alibaba.dubbo.config.annotation.Reference;
import com.alibaba.fastjson.JSON;
import com.qingcheng.pojo.goods.Sku;
import com.qingcheng.service.goods.SkuService;
import com.qingcheng.service.goods.SpuService;
import org.elasticsearch.action.bulk.BulkRequest;
import org.elasticsearch.action.index.IndexRequest;
import org.elasticsearch.action.index.IndexResponse;
import org.elasticsearch.action.search.SearchRequest;
import org.elasticsearch.client.RequestOptions;
import org.elasticsearch.client.RestHighLevelClient;
import org.elasticsearch.index.query.QueryBuilders;
import org.elasticsearch.index.reindex.DeleteByQueryRequest;
import org.elasticsearch.rest.RestStatus;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.amqp.core.Message;
import org.springframework.amqp.core.MessageListener;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;

import java.io.IOException;
import java.io.UnsupportedEncodingException;
import java.util.HashMap;
import java.util.List;
import java.util.Map;

/**
 * @Author:boy
 * @Date:2019/7/25
 * @description:
 * @ModifiedBy:
 */
@Component
public class GoodsAddIndexConsumer implements MessageListener {
    private Logger logger= LoggerFactory.getLogger(GoodsAddIndexConsumer.class);
    @Autowired
    private RestHighLevelClient restHighLevelClient;
    @Reference
    private SkuService skuService;
    public void onMessage(Message message) {
        try {
       String id=new String(message.getBody(),"utf-8");
       this.addIndex(id);
        } catch (UnsupportedEncodingException e) {
            logger.error("消息编码异常",e);
        }
    }

    /**
     * 根据spuid批量添加sku到索引库
     * @param id spuId
     */
    private void addIndex(String id) {
        //封装查询条件
        Map<String,Object> searchMap=new HashMap<String, Object>();
        searchMap.put("spuId",id);
        //根据spuId查询sku列表
        List<Sku> skuList = skuService.findList(searchMap);
        //批处理请求对象
        BulkRequest bulkRequest=new BulkRequest();
        //封装数据
        for (Sku sku : skuList) {
            //组装请求
            IndexRequest indexRequest=new IndexRequest("sku","doc");
            indexRequest.id(sku.getId());
            Map<String,Object> map=new HashMap<String,Object>();
            map.put("name",sku.getName());
            map.put("price",sku.getPrice());
            map.put("spuId",sku.getSpuId());
            map.put("saleNum",sku.getSaleNum());
            map.put("categoryName",sku.getCategoryName());
            map.put("commentNum",sku.getCommentNum());
            Map spec =JSON.parseObject(sku.getSpec(), Map.class);
            map.put("spec",spec);
            map.put("brandName",sku.getBrandName());
            map.put("image",sku.getImage());
            map.put("createTime",sku.getCreateTime());
            indexRequest.source(map);
            bulkRequest.add(indexRequest);
        }
        try {
            //批量更新
            restHighLevelClient.bulk(bulkRequest,RequestOptions.DEFAULT);
        } catch (IOException e) {
            logger.error("索引库批处理异常",e);
        }
    }
}
